package com.zxy.ziems.common.mqtt.template;

import cn.hutool.core.util.RandomUtil;
import com.zxy.btp.common.util.JsonUtils;
import com.zxy.ziems.common.mqtt.MqttCallbackImpl;
import com.zxy.ziems.common.mqtt.MqttMessageEncoder;
import com.zxy.ziems.common.mqtt.annotation.MqttTopic;
import com.zxy.ziems.common.mqtt.pojo.MqttMessagePayload;
import com.zxy.ziems.common.mqtt.pojo.Pattern;
import com.zxy.ziems.common.mqtt.pojo.SubscriptTopic;
import com.zxy.ziems.common.utils.IPUtils;
import com.zxy.ziems.common.mqtt.config.MqttClientProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author chdzq
 * @create 2024/4/8
 */
@Slf4j
@Component
@ConditionalOnBean(MqttClientProperties.class)
public class MqttClientTemplate {
    private final MqttClient mqttClient;

    public MqttClientTemplate(@Autowired MqttClientProperties properties,
                              ApplicationContext applicationContext,
                              Environment environment) throws Exception {
        //得到所有使用@Topic注解的类
        List<SubscriptTopic> topicMap = new ArrayList<SubscriptTopic>();
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(MqttTopic.class);
        for (String className : beansWithAnnotation.keySet()) {
            Class<?> classByteCode = beansWithAnnotation.get(className).getClass();
            //获取类的注解属性
            MqttTopic annotation = AnnotationUtils.findAnnotation(classByteCode, MqttTopic.class);
            String topic = annotation.topic();
            String realTopic = environment.resolvePlaceholders(topic);
            int qos = annotation.qos();
            Pattern patten = annotation.patten();
            String group = annotation.group();
            String realGroup = environment.resolvePlaceholders(group);

            String subTopic = realTopic;
            if (patten == Pattern.SHARE) {
                subTopic = "$share/" + realGroup + "/" + realTopic;
            } else if (patten == Pattern.QUEUE) {
                subTopic = "$queue/" + topic;
            }
            topicMap.add(new SubscriptTopic(realTopic, subTopic, patten, qos, (IMqttMessageListener) applicationContext.getBean(classByteCode)));
        }

        MqttClient mqttClient = new MqttClient(
                properties.getHost(),
                buildClientId(properties.getClientId()),
                new MemoryPersistence());

        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(properties.getUsername());
        options.setPassword(properties.getPassword().toCharArray());
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);

        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*10秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(5);
        options.setConnectionTimeout(0);
        options.setMqttVersion(4);
        options.setAutomaticReconnect(true);
        options.setMaxInflight(1000);

        mqttClient.setCallback(new MqttCallbackImpl(topicMap, mqttClient, options));
        mqttClient.connect(options);
        this.mqttClient = mqttClient;
    }

    public void publish(String topic, int qos, String message) {
        log.debug("[MQTT]发送消息：topic:{}, message:{}", topic, message);
        try {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setRetained(false);
            mqttMessage.setPayload(message.getBytes());
            mqttClient.publish(topic, mqttMessage);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 发布标准的数据
     * @param topic
     * @param entity
     * @param <T>
     */
    public <T> void publishPayload(String topic, T entity) {
        MqttMessagePayload<T> payload = new MqttMessagePayload<>(entity, LocalDateTime.now());
        publish(topic, payload);
    }
    public <T> void publish(String topic, T entity) {
        publish(topic, entity, JsonUtils::toJSONString);
    }

    public <T> void publish(String topic, T entity, MqttMessageEncoder<T> encoder) {
        publish(topic, 1, encoder.mqttEncodeMessage(entity));
    }

    private String buildClientId(String clientId) throws Exception {
        return clientId.concat("@").concat(IPUtils.getLocalInetAddress().getHostAddress().concat(":").concat(String.valueOf(RandomUtil.randomInt(3))));
    }
}
